package com.wchen.project.zk.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Desc:   使用 Curator 实现分布式 Barrier
 * @author: cww
 * @DateTime: 2020-01-25 11:27
 */

public class ZkDistBarrierUseCurator {
    static String barrier_path = "/curator_recipes_barrier_path";
    static DistributedBarrier barrier;

    public static void main(String[] args) throws Exception{
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("home:2181")
                                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                                .build();
                        client.start();
                        barrier = new DistributedBarrier(client, barrier_path);
                        System.out.println(Thread.currentThread().getName() + "号barrier设置");
                        barrier.setBarrier();
                        // 等待 barrier 的释放
                        barrier.waitOnBarrier();
                        System.out.println("启动...");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        Thread.sleep(2000);
        // 通过 removeBarrier 方法来释放 Barrier ，同时触发所有等待该 Barrier 的 5 个线程同时进行各自的业务逻辑
        barrier.removeBarrier();
    }
}

/**
 * Curator 提供的另一种线程自发触发 Barrier 释放的模式
 */
class ZkDistBarrierUseCurator2 {
    static String barrier_path = "/curator_recipes_barrier_path";

    public static void main(String[] args) {
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        CuratorFramework client = CuratorFrameworkFactory.builder()
                                .connectString("home:2181")
                                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                                .build();
                        client.start();
                        // 成员数阈值 5
                        DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, barrier_path, 5);
                        Thread.sleep(Math.round(Math.random() * 3000));
                        System.out.println(Thread.currentThread().getName() + "号进入 barrier");
                        // 每个 Barrier 的参与者都会在 enter 方法之后进行等待，当所有成员都到达后，将被同时出发进入
                        barrier.enter();
                        System.out.println("启动...");
                        Thread.sleep(Math.round(Math.random() * 3000));
                        // 调用 leave 方法，则线程会再次等待，此时处于准备退出状态，一旦准备退出 Barrier 的成员数达到
                        // 5 个后，所有成员同样会被同时触发退出
                        barrier.leave();
                        System.out.println("退出...");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

/**
 * 使用 CyclicBarrier 模拟一个赛跑比赛
 * 在同一个 JVM 中，使用 CyclicBarrier 就可以解决诸如此类的多线程同步问题
 */
class DemoCyclicBarrier {
    public static CyclicBarrier barrier = new CyclicBarrier(3);

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(new Thread(new Runner("1 号选手")));
        executor.submit(new Thread(new Runner("2 号选手")));
        executor.submit(new Thread(new Runner("3 号选手")));
        executor.shutdown();
    }
}

class Runner implements Runnable {
    private String name;

    public Runner(String name) {
        this.name = name;
    }

    @Override
    public void run() {
        System.out.println(name + "准备好了。");
        try {
            DemoCyclicBarrier.barrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(name + "起跑！");
    }
}
